package com.guchenbo.gpark.gd

/**
 * @author guchenbo
 * @date 2022/3/22
 */
class PairGDFunction[K, V](gd: GD[(K, V)]) {

  def groupByKey(): GD[(K, Seq[V])] = {
    val f = (seq: Seq[(K, V)]) => seq.groupBy(_._1).map(a => (a._1, a._2.map(_._2))).toSeq
    new MapGD(gd, f)
  }

  def reduceByKey(func: (V, V) => V): GD[(K, V)] = {
    val f = (seq: Seq[(K, V)]) =>
      seq.groupBy(_._1).map(a => (a._1, a._2.map(_._2).reduce(func))).toSeq
    new MapGD(gd, f)
  }

  def foldByKey(zero: V)(func: (V, V) => V): GD[(K, V)] = {
    val f = (seq: Seq[(K, V)]) =>
      seq.groupBy(_._1).map(a => (a._1, a._2.map(_._2).fold(zero)(func))).toSeq
    new MapGD(gd, f)
  }
}
